/*
 * Copyright (c) 2022-2023 cetcclout Co., Ltd All rights reserved.
 * Sensor Adapter is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
package com.cetccloud.cimp.adapter.server.mqtt.publish;

import cn.hutool.json.JSONUtil;
import com.cetccloud.cimp.adapter.cache.ClientDeviceCache;
import com.cetccloud.cimp.adapter.cache.MqttCenter;
import com.cetccloud.cimp.adapter.config.ApplicationProperties;
import com.cetccloud.cimp.adapter.datasource.redis.RedisService;
import com.cetccloud.cimp.adapter.datasource.redis.impl.RedisServiceImpl;
import com.cetccloud.cimp.adapter.handler.DataContextValue;
import com.cetccloud.cimp.common.util.SensorStringUtils;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

/**
 * @author
 * @apiNote 每个设备创建一个mqtt客户端链接
 * @date: 2022/11/4 11:23
 */
@Component
@Async
public class MqttDtuIdClientService {
//    @Value("${application.mqtt.broker}")
//    private String broker;
//    @Value("${application.mqtt.pwd}")
//    private String pwd;
    @Value("${spring.profiles.active}")
    private String active;
    int qos = 2;
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(MqttDtuIdClientService.class);

    /**
     * 所有设备配置一个连接前缀 clientId
     */
    private final String clientId = SensorStringUtils.SUBSCRIBE_CLIENT + SensorStringUtils.getHostName().toLowerCase() + "-";

    @Autowired
    private RedisService redisService;
    @Autowired
    ApplicationProperties applicationProperties;


    /**
     * 创建MQTT链接客户端
     */
    public synchronized MqttClient MakeClient(String dtuId){
        MqttClient mqttDeviceClient = MqttCenter.findMqttClient(dtuId);
        if(null == mqttDeviceClient){
            synchronized (MqttClient.class) {
                if(null == mqttDeviceClient) {
                    MemoryPersistence persistence = new MemoryPersistence();
                    try {
                        mqttDeviceClient = new MqttClient(applicationProperties.getMqtt().getBroker(), clientId + active + "-" + dtuId, persistence);
                        MqttConnectOptions connOpts = new MqttConnectOptions();
                        connOpts.setCleanSession(true);
                        connOpts.setUserName("edge");
                        connOpts.setPassword(applicationProperties.getMqtt().getPwd().toCharArray());
                        // 设置超时时间
                        connOpts.setConnectionTimeout(10);
                        connOpts.setKeepAliveInterval(3600);
                        connOpts.setAutomaticReconnect(true);
                        connOpts.setMaxInflight(1000);
                        mqttDeviceClient.connect(connOpts);
                        MqttCenter.addMqttClient(dtuId,mqttDeviceClient);
                        logger.info(mqttDeviceClient.getClientId() + "::mqtt连接初始化成功");
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                }
            }
        }else if(!mqttDeviceClient.isConnected()){
            logger.info("mqtt链接失效，重连！！！");
            try {
                mqttDeviceClient.reconnect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

        return mqttDeviceClient;
    }

    /**
     * 发布采集数据
     */
    public void SendMsgSecond(String  msgDataPackage) throws MqttException {
        String dtuId = JSONUtil.parseObj(msgDataPackage).getStr("deviceId");
        MqttClient sampleClient = MqttCenter.findMqttClient(dtuId);
        String topicPre = applicationProperties.getMqtt().getTopic();
        String topic = topicPre + dtuId;
        logger.info("topic:"+ topic);
        String content = msgDataPackage;
        if(sampleClient!=null){
            if(!sampleClient.isConnected()){
                sampleClient.reconnect();
            }
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            sampleClient.publish(topic, message);
            logger.warn("E2MIS_adapter_send: {} <-> {}",dtuId, LocalDateTime.now());
            logger.info("发布采集数据：{}",message.toString());
        }else {
            if(redisService.hasKey(String.format(RedisServiceImpl.JR_DEVICE_MODEL_KEY, dtuId))) {
                this.MakeClient(dtuId);
                logger.info("重新注册MQTT链接发送数据 ： " + dtuId);
                //非设备正常连接 不记录
                redisService.sSet(DataContextValue.DTU_STATUS_KEY, dtuId);
                sampleClient = MqttCenter.findMqttClient(dtuId);
                MqttMessage message = new MqttMessage(content.getBytes());
                message.setQos(qos);
                sampleClient.publish(topic, message);
                logger.warn("E2MIS_adapter_send: {} <-> {}",dtuId, LocalDateTime.now());
                logger.info("重新发布采集数据：{}",message.toString());
            }else{
                logger.info("无有效MQTT链接，数据未发送<==" + content);
                ClientDeviceCache.getChannel(dtuId).close();
                ClientDeviceCache.removeMap(dtuId);
            }
        }
    }

    /**
     * 关闭MQTT链接客户端
     */
    public void CloseMqttClient(String dtuId) throws MqttException {
        MqttClient sampleClient=MqttCenter.findMqttClient(dtuId);
        if(sampleClient.isConnected()){
            sampleClient.disconnect();
            sampleClient.close();
        }
        MqttCenter.removeMqttClient(dtuId);
        logger.info(sampleClient.getClientId()+"::Mqtt连接关闭");
    }

}
